package com.atguigu.hadoop.examples.mapreduce.terasort;


import java.io.FileNotFoundException;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileAlreadyExistsException;
import org.apache.hadoop.mapred.InvalidJobConfException;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.security.TokenCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TeraOutputFormat extends FileOutputFormat<Text, Text>
{
    private static final Logger LOG = LoggerFactory.getLogger(TeraOutputFormat.class)
            ;
    private OutputCommitter committer = null;

    static void setFinalSync(JobContext job, boolean newValue)
    {
        job.getConfiguration().setBoolean(TeraSortConfigKeys.FINAL_SYNC_ATTRIBUTE
                .key(), newValue);
    }

    public static boolean getFinalSync(JobContext job)
    {
        return job.getConfiguration().getBoolean(TeraSortConfigKeys.FINAL_SYNC_ATTRIBUTE
                .key(), true);
    }

    public void checkOutputSpecs(JobContext job)
            throws InvalidJobConfException, IOException
    {
        Path outDir = getOutputPath(job);
        if (outDir == null) {
            throw new InvalidJobConfException("Output directory not set in JobConf.");
        }

        Configuration jobConf = job.getConfiguration();

        TokenCache.obtainTokensForNamenodes(job.getCredentials(), new Path[] { outDir }, jobConf);

        FileSystem fs = outDir.getFileSystem(jobConf);
        try
        {
            FileStatus[] outDirKids = fs.listStatus(outDir);
            boolean empty = false;
            if ((outDirKids != null) && (outDirKids.length == 1)) {
                FileStatus st = outDirKids[0];
                String fname = st.getPath().getName();

                empty = (!st
                        .isDirectory()) && ("_partition.lst".equals(fname));
            }
            if ((TeraSort.getUseSimplePartitioner(job)) || (!empty))
                throw new FileAlreadyExistsException("Output directory " + outDir + " already exists");
        }
        catch (FileNotFoundException localFileNotFoundException)
        {
        }
    }

    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext job) throws IOException
    {
        Path file = getDefaultWorkFile(job, "");
        FileSystem fs = file.getFileSystem(job.getConfiguration());
        FSDataOutputStream fileOut = fs.create(file);
        return new TeraRecordWriter(fileOut, job);
    }

    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException
    {
        if (this.committer == null) {
            Path output = getOutputPath(context);
            this.committer = new FileOutputCommitter(output, context);
        }
        return this.committer;
    }

    static class TeraRecordWriter extends RecordWriter<Text, Text>
    {
        private boolean finalSync = false;
        private FSDataOutputStream out;

        public TeraRecordWriter(FSDataOutputStream out, JobContext job)
        {
            this.finalSync = TeraOutputFormat.getFinalSync(job);
            this.out = out;
        }

        public synchronized void write(Text key, Text value) throws IOException
        {
            this.out.write(key.getBytes(), 0, key.getLength());
            this.out.write(value.getBytes(), 0, value.getLength());
        }

        public void close(TaskAttemptContext context) throws IOException {
            if (this.finalSync) {
                try {
                    this.out.hsync();
                }
                catch (UnsupportedOperationException e)
                {
                    TeraOutputFormat.LOG.info("Operation hsync is not supported so far on path with erasure code policy set");
                }
            }

            this.out.close();
        }
    }
}